import log4js from "log4js"
import geoip from 'geoip-lite'
import axios, { Axios, AxiosInstance, AxiosResponse } from "axios"
import dayjs from "dayjs"
import { MongoDBClient } from "../db"
const logger = log4js.getLogger()












export class ETLClient {
    extractorQueue: ProxyInterface[] = []
    loaderQueue: ProxyInterface[] = []
    constructor(
        private extractor: ExtractClientInterface,
        private transformer: TransformerClientInterface,
        private concurrency: number,
        private loader: LoadClientInterface,
    ) { }

    async extract() {
        while (true) {
            const datas = this.extractor.extract()
            for await (const item of datas) {
                this.extractorQueue.push(item)
            }
            await sleep(3600)
        }
    }

    async transform(): Promise<ProxyInterface> {
        while (true) {
            const item = this.extractorQueue.shift()
            if (!item) {
                await sleep(1)
                continue
            }
            let newItem: ProxyInterface | undefined = undefined
            try {
                if (item.protocol === 'socks4') continue
                newItem = await this.transformer.transform(item)
            } catch (err: any) {
                logger.error(`tranform error while handling data: ${JSON.stringify(item)}: ${err}`)
                continue
            }

            if (!newItem) continue
            this.loaderQueue.push(newItem)
        }

    }

    async load() {
        while (true) {
            const item = this.loaderQueue.shift()
            if (!item) {
                await sleep(1)
                continue
            }

            try {
                await this.loader.load(item)
            } catch (err: any) {
                logger.error(`load error while handling data: ${JSON.stringify(item)}: ${err}`)
                continue
            }
        }
    }

    etl() {
        this.extract()
        for (let i = 0; i < this.concurrency; i += 1) {
            this.transform()
        }
        this.load()
    }
}




export interface ProxyInterface {
    _id?: string
    ip: string
    port: number
    protocol?: string
    country?: string
    city?: string
    reachable?: boolean
    rtt?: number
    scope?: string[]
    created?: number
    updated?: number
    failed_count?: number
}

export interface ExtractClientInterface {
    extract(): AsyncGenerator<ProxyInterface, void, any>
}

export interface TransformerClientInterface {
    transform(data: ProxyInterface): Promise<ProxyInterface>
}

export interface LoadClientInterface {
    load(data: ProxyInterface): Promise<void>
}





export class TransformProtocol implements TransformerClientInterface {

    async transform(data: ProxyInterface): Promise<ProxyInterface> {
        // const axiosArr: Promise<AxiosResponse<never>>[] = []
        // const protocols = ['http', 'socks5']
        // const scope:string[] = []
        // data.protocol = '1'
        // data.scope = ['ada']
        // for (let i = 0; i < 2; i++) {
        //     const paramUrl = `${protocols[i]}://${data.ip}:${data.port}`
        //     try {
        //         axiosArr.push(axios.get(`http://172.16.0.90:8080/api/v1/proxy/validate`, { params: { url: paramUrl, target: "https://baidu.com" } }))
        //         axiosArr.push(axios.get(`http://172.16.0.90:8080/api/v1/proxy/validate`, { params: { url: paramUrl, target: "https://google.com" } }))
        //     } catch (error) {
        //         logger.error(`transformProtocol transform request ${error}`)
        //         data.reachable = false
        //         break
        //     }
        // }

        // if (axiosArr.length !== 4) return data
        // const axiosRes: AxiosResponse<never>[] = await Promise.all(axiosArr)
        // data.reachable = false
        // for (let i = 0; i < axiosArr.length; i++) {
        //     const res = axiosRes[i]
        //     const status = (res.data as any).status
        //     if (!status) continue
            // data.reachable = true
        //     switch (i) {
        //         case 0:
        //             data.protocol = 'http'
        //             dscope.push(`https://www.baidu.com`)
        //             break
        //         case 1:
        //             data.protocol = 'http'
        //             scope.push(`https://www.google.com`)
        //             break
        //         case 2:
        //             data.protocol = 'socks5'
        //             scope.push(`https://www.baidu.com`)
        //             break
        //         case 3:
        //             data.protocol = 'socks5'
        //             scope.push(`https://www.google.com`)
        //     }
        // }
        // data.scope = scope
        const arr:string[]  = []
        arr.push(`aaaa`)
        arr.push(`bbbb`)
        data.scope = arr
        data.protocol = 'LiasouX'
        return data
    }
}



export class TransformRTT implements TransformerClientInterface {

    async transform(data: ProxyInterface): Promise<ProxyInterface> {
        // const url = `http://172.16.0.90:8080/api/v1/ping/tcp/${data.ip}:${data.port}`
        // let res: AxiosResponse
        // try {
        //     res = await axios.get(url)
        // } catch (error) {
        //     logger.error(`TransformRTT transform axios request ${error}`)
        //     data.reachable = false
        //     data.rtt = 0
        //     return data
        // }
        // const dataRtt = res.data as any
        // const rtt: number = dataRtt.rtt
        data.rtt = 12
        return data
    }

}



export class TransformMessage implements TransformerClientInterface {
    async transform(data: ProxyInterface): Promise<ProxyInterface> {
        const geo = geoip.lookup(data.ip)
        data.country = geo?.country
        data.city = geo?.city
        return data
    }
}


export class UpdateExtract implements ExtractClientInterface {

    constructor(
        private client: MongoDBClient<ProxyInterface>
    ) { }

    async *extract(): AsyncGenerator<ProxyInterface, void, any> {
        // const datas = await this.client.findAll()
        // for (const item of datas) {
        //     yield item
        // }
    }
}


export class LoadClient implements LoadClientInterface {

    constructor(
        protected client: MongoDBClient<ProxyInterface>
    ) { }

    async load(data: ProxyInterface): Promise<void> {
        data._id = `${data.ip}:${data.port}`
        console.log(data)
        // try {
        //     await this.client.insertOne(data)
        // } catch (error) {
        //     logger.error(`LoadClient load inserOne ${error}`)
        // }
    }
}

export class MultiExtractClient implements ExtractClientInterface {
    constructor(
        private extractors: ExtractClientInterface[]
    ) { }
    async * extract(): AsyncGenerator<any, void, ProxyInterface> {
        for (const extractor of this.extractors) {
            try {
                const datas = extractor.extract()
                for await (const item of datas) {
                    yield item
                }
            } catch (err) {
                logger.error(`extract error while handling: ${err}`)
            }
        }
    }
}

export class TransformerPipeline implements TransformerClientInterface {
    constructor(
        private transformers: TransformerClientInterface[]
    ) { }

    async transform(data: ProxyInterface): Promise<ProxyInterface> {
        let item: ProxyInterface = { ...data }

        for (const transformer of this.transformers) {
            try {
                item = await transformer.transform(item)
            } catch (err) {
                logger.error(`tranform error while handling data: ${JSON.stringify(item)}: ${err}`)
            }
        }
        return item
    }
}

// export class ETLClient {
//     constructor(
//         private extractor: ExtractClientInterface,
//         private transformer: TransformerClientInterface,
//         private loader: LoadClientInterface,
//     ) { }

//     async etl() {
//         const datas = this.extractor.extract()
//         for await (const item of datas) {
//             let newItem: ProxyInterface | undefined = undefined
//             try {
//                 if (item.protocol === 'socks4') continue
//                 item.reachable = true
//                 newItem = await this.transformer.transform(item)
//             } catch (err: any) {
//                 logger.error(`tranform error while handling data: ${JSON.stringify(item)}: ${err}`)
//                 continue
//             }

//             if (!newItem) continue

//             try {
//                 await this.loader.load(newItem)
//             } catch (err: any) {
//                 logger.error(`load error while handling data: ${JSON.stringify(newItem)}: ${err}`)
//                 continue
//             }
//         }
//     }
// }



async function sleep(time: number) {
    return new Promise((resolve, reject) => {
        setTimeout(() => {
            resolve(1)
        }, 1000 * time);
    })
}